---
title: Fork/Join
---

线程池、Future、CompletableFuture 和 CompletionService 这些工具类都是站在任务的视角来解决并发任务。而不是纠缠在线程之间如何协作的细节上，

**对于简单的并行任务，使用线程池+Future 的方案来解决；如果任务之间有聚合关系，无论是 AND 聚合还是 OR 聚合，都可以通过 CompletableFuture 来解决；而批量的并行任务，则可以通过 CompletionService 来解决。**

并发编程可以分成三个层面的问题，分别是分工、协作和互斥，当你关注于任务的时候，就会发现你的视角已经从并发编程的细节中跳出来了，可以类比现实世界中的分工，所以把 线程池、Future、CompletableFuture 和 CompletionService 都列到了分工里面。

![47f3e1e8834c99d9a1933fb496ffde2d](https://img.wkq.pub/pic/47f3e1e8834c99d9a1933fb496ffde2d.webp)

从上到下，依次为简单并行任务、聚合任务和批量并行任务示意图。



上面提到的简单并行、聚合、批量并行这三种任务模型，基本上能够覆盖日常工作中的并发场景了，但是还不够全面，因为还有一种“分治”的任务模型没有覆盖到。



分治，顾名思义，即分而治之，是一种解决复杂问题的思维方法和模式；

具体来讲，指的是**把一个复杂的问题分解成多个相似的子问题，然后再把子问题分解成更小的子问题，直到子问题简单到可以直接求解**。理论上讲，解决每一个问题都对应着一个任务，所以对于问题的分治，实际上就是对于任务的分治。



Java 并发包里提供了一种叫做 Fork/Join 的并行计算框架，就是用来支持分治这种模型的。



## 分治任务模型

分治任务模型可以分成两个阶段：一个阶段是**任务分解**，也就是将任务迭代地分解为子任务，直至子任务可以直接计算出结果；另一个阶段是**结果合并**，即逐层合并子任务的执行结果，直至获得最终结果。

![d2649d8db8e5642703aa5563d76eb86a](https://img.wkq.pub/pic/d2649d8db8e5642703aa5563d76eb86a.webp)

在这个分治模型里，任务和分解后的子任务具有相似性，这种相似性往往体现在任务和子任务的算法是相同的，但是计算的数据规模是不同的。

具备这种相似性的问题，我们往往都采用递归算法。

## Fork/Join 的使用

Fork/Join 是一个并行计算的框架，主要就是用来支持分治任务模型的，这个计算框架里的** Fork 对应的是分治任务模型里的任务分解，Join 对应的是结果合并**。Fork/Join 计算框架主要包含两部分，一部分是分治任务的线程池 **ForkJoinPool** ，另一部分是分治任务 **ForkJoinTask**。这两部分的关系类似于 ThreadPoolExecutor 和 Runnable 的关系，都可以理解为提交任务到线程池，只不过分治任务有自己独特类型的 ForkJoinTask。



ForkJoinTask 是一个抽象类，它的方法有很多，最核心的是 fork() 方法和 join() 方法，其中 fork() 方法会异步地执行一个子任务，而 join() 方法则会阻塞当前线程来等待子任务执行的结果。

ForkJoinTask 有两个子类 ———— RecursiveAction 和 RecursiveTask,它们都是用递归的方式来处理分治任务的。这两个子类都定义了抽象方法 compute() ，区别是 RecursiveAction 定义的 compute() 没有返回值，而 RecursiveTask 定义 compute() 方法是有返回值。这两个类也是抽象类，在实现的时候需要定义子类去扩展。

```java title='使用 Fork/Join 计算斐波那契数列'
public class Test {
    public static void main(String[] args) {
        //创建分治任务线程池
        ForkJoinPool fjp = new ForkJoinPool(4);
        //创建分治任务
        Fibonacci fibonacci = new Fibonacci(30);
        //启动分治任务
        Integer result = fjp.invoke(fibonacci);
        System.out.println("result = " + result);
    }
    static class Fibonacci extends RecursiveTask<Integer> {
        final int n;

        Fibonacci(int n) {
            this.n = n;
        }
        @Override
        protected Integer compute() {
            if (n <= 1) {
                return n;
            }
            Fibonacci f1 = new Fibonacci(n - 1);
            //创建子任务
            f1.fork();
            Fibonacci f2 = new Fibonacci(n - 2);
            //等待子任务结果，并合并结果
            return f2.compute() + f1.join();
        }
    }
}
```

## ForkJoinPool 工作原理

Fork/Join 并行计算的核心组件时 ForkJoinPool。

ThreadPoolExecutor 本质上是一个生产者 —  消费者模式的实现，内部有一个任务队列，这个任务队列是生产者和消费者通信的媒介；ThreadPoolExecutor 可以有多个工作线程，但是这些工作线程都共享一个任务队列。

ForkJoinPool 本质上也是一个生产者 — 消费者的实现，但是更加智能，ThreadPoolExecutor 内部只有一个工作队列，而 ForkJoinPool 内部有多个任务队列，当我们通过 ForkJoinPool 的 invoke () 或者 submit() 方法提交任务时，ForkJoinPool 根据一定的路由规则把任务提交到一个任务队列中，如果任务在执行过程中会创建出子任务，那么子任务会提交到工作线程对应的任务队列中。



如果工作线程对应的任务队列空了，那可以窃取其它工作任务队列里的任务，如此一来，所有的工作线程都不会闲下来了。



![e75988bd5a79652d8325ca63fcd55131](https://img.wkq.pub/pic/e75988bd5a79652d8325ca63fcd55131.webp)

## 

```java title='模拟MapReduce统计单词数量'
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

/**
 * 学习 MapReduce 有一个入门程序，统计一个文件里面每个单词的数量，下面我们来看看如何用 Fork/Join 并行计算框架来实现。我们可以先用二分法递归地将一个文件拆分成更小的文件，直到文件里只有一行数据，然后统计这一行数据里单词的数量，最后再逐级汇总结果，你可以对照前面的简版分治任务模型图来理解这个过程。思路有了，我们马上来实现。下面的示例程序用一个字符串数组 String[] fc 来模拟文件内容，fc 里面的元素与文件里面的行数据一一对应。关键的代码在 compute() 这个方法里面，这是一个递归方法，前半部分数据 fork 一个递归任务去处理（关键代码 mr1.fork()），后半部分数据则在当前任务中递归处理（mr2.compute()）。
 */
public class Test {
    public static void main(String[] args) {
        String[] fc = {"hello world", "hello me", "hello fork", "hello join", "fork join in world"};
        //创建分治任务线程池
        ForkJoinPool fjp = new ForkJoinPool(3);
        //创建分治任务
        var mr = new MR(fc, 0, fc.length);
        Map<String, Integer> map = fjp.invoke(mr);
        //启动分治任务
        System.out.println("result = " + map);
    }

    static class MR extends RecursiveTask<Map<String, Integer>> {
        private String[] fc;
        private int start, end;

        public MR(String[] fc, int start, int end) {
            this.fc = fc;
            this.start = start;
            this.end = end;
        }


        @Override
        protected Map<String, Integer> compute() {
            if (end - start == 1) {
                return calc(fc[start]);
            } else {
                int mid = (start + end) / 2;
                MR mr1 = new MR(fc, start, mid);
                mr1.fork();
                MR mr2 = new MR(fc, mid, end);
                return merget(mr1.join(), mr2.compute());
            }
        }

        private Map<String, Integer> merget(Map<String, Integer> r1, Map<String, Integer> r2) {
            Map<String, Integer> result = new HashMap<>();
            result.putAll(r1);
            //合并结果
            r2.forEach((k,v)->{
                Integer c = result.get(k);
                if (c != null) {
                    result.put(k, c + v);
                } else {
                    result.put(k, v);
                }
            });
            return result;
        }

        //统计单词的数量
        private Map<String, Integer> calc(String line) {
            Map<String, Integer> result = new HashMap<>();
            //分割单词
            String[] words = line.split("\\s+");
            for (String word : words) {
                Integer v = result.get(word);
                if (v != null) {
                    result.put(word, v + 1);
                } else {
                    result.put(word, 1);
                }
            }
            return result;
        }
    }
}

```

## 总结

Fork/Join 并行计算框架主要解决的时分治任务。分治的核心思想是分而治之；将一个大的任务拆分成小的子任务去解决，然后再把子任务的结果聚合起来从而得到最终结果。

Fork/Join 并行计算框架的核心组件是 ForkJoinPool。ForkJoinPool 支持任务窃取机制，能够让所有线程的工作量基本均衡，不会出现有的线程很忙，而有的线程很闲的情况，所以性能很好。Java 1.8 里提供的并行流也是以 ForkJoinPool 为基础的。默认的并行流计算都共享一个 ForkJoinPool，这个共享的 ForkJoinPool 默认的线程数是 CPU 的核数；如果所有并行流计算都是 CPU 密集型计算的话，完全没有问题，但是如果存在 I/O 密集型的并行流运算，那么很可能因为一个很慢的 I/O 计算而拖慢整个系统的性能。**所以建议用不同的 ForkJoinPool 执行不同类型的计算任务。**

